package net.xo68.destiny.starter.manager.sched.impl;

import lombok.extern.slf4j.Slf4j;
import net.xo68.destiny.contract.common.DistributedLockKeys;
import net.xo68.destiny.contract.common.SystemConstants;
import net.xo68.destiny.core.lock.DistributedLock;
import net.xo68.destiny.core.lock.DistributedLockManager;
import net.xo68.destiny.core.util.IpUtil;
import net.xo68.destiny.core.util.ThreadUtils;
import net.xo68.destiny.core.util.TimeRecorder;
import net.xo68.destiny.starter.common.SchedulerCommonProperties;
import net.xo68.destiny.starter.manager.SchedulerManagerProperties;
import net.xo68.destiny.starter.manager.sched.JobTriggerObserver;
import net.xo68.destiny.starter.manager.sched.JobTriggerObserverProvider;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Component;

import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 任务触发观察者实现
 */
@Slf4j
@Component
public class JobTriggerObserverImpl implements JobTriggerObserver {

    /**
     * 初始化
     */
    public static final int STATE_INIT=0;
    /**
     * 已经启动
     */
    public static final int STATE_STARTED=1;
    /**
     * 关闭的
     */
    public static final int STATE_SHUTDOWN=2;

    /**
     * 状态
     */
    private final AtomicInteger runState=new AtomicInteger(STATE_INIT);

    private final ScheduledExecutorService executorService;


    private final Environment environment;

    private final SchedulerCommonProperties schedulerCommonProperties;

    private final SchedulerManagerProperties schedulerManagerProperties;

    private final DistributedLockManager distributedLockManager;

    private final JobTriggerObserverProvider jobTriggerObserverProvider;
    /**
     * 最大观察数量
     */
    private int maxObserveSize=1000;

    public JobTriggerObserverImpl(Environment environment, SchedulerCommonProperties schedulerCommonProperties, SchedulerManagerProperties schedulerManagerProperties, DistributedLockManager distributedLockManager, JobTriggerObserverProvider jobTriggerObserverProvider) {
        this.environment = environment;
        this.schedulerCommonProperties = schedulerCommonProperties;
        this.schedulerManagerProperties = schedulerManagerProperties;
        this.distributedLockManager = distributedLockManager;
        this.jobTriggerObserverProvider = jobTriggerObserverProvider;
        this.executorService = ThreadUtils.newSingleThreadScheduled("JobTriggerObserver");
        this.maxObserveSize=this.schedulerManagerProperties.getTriggerObserverMaxSize();
    }


    @Override
    public void run() {
        try {
            String lockHold= IpUtil.getIpPort(schedulerCommonProperties.getNodeRealIp(),environment.getProperty(SystemConstants.CONFIG_SERVER_PORT));
            DistributedLock lock = distributedLockManager.tryLock(DistributedLockKeys.JOB_TRIGGER_OBSERVER, lockHold, schedulerManagerProperties.getTriggerObserverTimeout());
            if(lock!=null){
                //LocalDateTime maxTime=TimeUtils.nowLocalDateTime().plusSeconds(this.schedulerManagerProperties.getTriggerObserverBufferTime());
                //log.debug("maxObserveSize: {}", maxObserveSize);
                long nowTime = System.currentTimeMillis();
                TimeRecorder timeRecorder=TimeRecorder.start();
                jobTriggerObserverProvider.oberveJob(this,this.schedulerManagerProperties.getTriggerObserverBufferTime(),maxObserveSize);
                timeRecorder.end();
                //动态调整调度数量，防止卡死
                if(timeRecorder.getRunMillis() < schedulerManagerProperties.getTriggerObserverTimeout()*1000*80/100){
                    if(maxObserveSize < this.schedulerManagerProperties.getTriggerObserverMaxSize()){
                        maxObserveSize++;
                    }
                }else {
                    if(maxObserveSize > this.schedulerManagerProperties.getTriggerObserverMaxSize()/10){
                        maxObserveSize--;
                    }
                }
                distributedLockManager.unLock(lock);
            }
        }catch (Exception e){
            log.error(e.getMessage(),e);
        }
    }

    @Override
    public void destroy() throws Exception {
        if(runState.compareAndSet(STATE_STARTED,STATE_SHUTDOWN)) {
            this.executorService.shutdown();
        }
    }

    @Override
    public void afterSingletonsInstantiated() {
        if(runState.compareAndSet(STATE_INIT,STATE_STARTED)){
            this.executorService.scheduleWithFixedDelay(this,SystemConstants.INITIAL_DELAY_SECOND,this.schedulerManagerProperties.getTriggerObserverPeriod(), TimeUnit.SECONDS);
        }
    }
}
